package com.bujunjie.study.realtime.dwd.db.app;

import com.bujunjie.study.realtime.common.base.BaseSQLApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.util.SQLUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * <p>读取评论信息</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/16 16:49
 */
public class DwdInteractionCommentInfo extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdInteractionCommentInfo().start(10019, 4, FlinkConstant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv) {
        this.readOdsDb(tableEnv, FlinkConstant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
        Table commentInfo = tableEnv.sqlQuery("SELECT\n" +
                "`data`['id'] AS id,\n" +
                "`data`['user_id'] AS userId,\n" +
                "`data`['sku_id'] AS skuId,\n" +
                "`data`['appraise'] AS appraise,\n" +
                "`data`['comment_text'] AS commentText,\n" +
                "`ts`,\n" +
                "`pt`\n" +
                "FROM topic_db\n" +
                "WHERE `table` = 'comment_info' \n" +
                "AND `type` = 'insert'");
        //  将查询出的 sql 转换成视图
        tableEnv.createTemporaryView("comment_info", commentInfo);

        //  读取 hbase 的字典表
        this.readBaseDic(tableEnv);
        //  采用 lookup join 将当前的评论表
        Table joinTable = tableEnv.sqlQuery("SELECT\n" +
                "    id,\n" +
                "    userId,\n" +
                "    skuId,\n" +
                "    appraise,\n" +
                "    dic.dic_name AS appraiseName,\n" +
                "    commentText,\n" +
                "    ts\n" +
                "FROM comment_info AS c\n" +
                "  JOIN base_dic FOR SYSTEM_TIME AS OF c.pt AS dic\n" +
                "    ON c.appraise = dic.dic_code");


        // 将关联的结果写到kafka主题中                    ---upsert kafka连接器
        //创建动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE " + FlinkConstant.TOPIC_DWD_INTERACTION_COMMENT_INFO + " (\n" +
                "    id string,\n" +
                "    userId string,\n" +
                "    skuId string,\n" +
                "    appraise string,\n" +
                "    appraiseName string,\n" +
                "    commentTxt string,\n" +
                "    ts bigint,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + SQLUtil.getUpsertKafkaDDL(FlinkConstant.TOPIC_DWD_INTERACTION_COMMENT_INFO));

        // 写入对应的表中
        joinTable.executeInsert(FlinkConstant.TOPIC_DWD_INTERACTION_COMMENT_INFO);

    }
}
